package com.aotain.project.apollo.bolt;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.log4j.Logger;
import org.apache.storm.Config;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.FailedException;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import com.aotain.hbase.HBaseRecordAdd;
import com.aotain.storm.AbstractBolt;
import com.aotain.storm.TupleHelpers;

/**
 * 
 * @ClassName: AbnStatisBolt 
 * @Description: TODO(这里用一句话描述这个类的作用) 
 * @author 程彬
 * @date 2016年4月28日 下午1:33:32 
 *
 */
public class AbnStatisBolt extends AbstractBolt{

	private static final long serialVersionUID = 1L;

//	String dip = null;
//	String dport = null;
//	String sip = null;
//	String sport = null;
//	String dateStr_d = null;
//	String dateStr_h = null;
//	String dateStr_hour = null;
//	String sourceAreaId = null;
//	String areaName = null;
//	String sourceAreaName = null;
//	String idcno = null;
//	String abnAreaMapKey = null;
//	String abnSessHMapKey = null;
//	String abnSessDMapKey = null;
//	String abnSessSourHMapKey = null;
//	String abnSessSourDMapKey = null;
//	String sourceCountry = null;
//	Long triplePV = null;
//	String abnormaltype = null;
//	Long streamOctets = 0L;
//	Long streamPacket = 0L;
	Map<String,Long> abnAreaMap = new HashMap<String,Long>();
	Map<String,Long> abnSessHMap = new HashMap<String,Long>();
	Map<String,Long> abnSessDOctetsMap = new HashMap<String,Long>();
	Map<String,Long> abnSessDPacketMap = new HashMap<String,Long>();
	Map<String,Long> abnSessSourHMap = new HashMap<String,Long>();
	Map<String,Long> abnSessSourDMap = new HashMap<String,Long>();
	Map<String,Long> abnSessionPVMap = new HashMap<String,Long>();


	

	@Override
	public void execute(Tuple tuple) {
		// TODO Auto-generated method stub
		try {
			if(TupleHelpers.isTickTuple(tuple)) {
				emitCountingData(collector);  	
				collector.ack(tuple);
			} else {
				countInLocal(tuple); 
				collector.ack(tuple);
			}
		} catch (Exception e) {
			// TODO Auto-generated catch block
			System.err.println(" AbnStatisBolt exception"+e.getMessage());
			Logger.getRootLogger().error("AbnStatisBolt exception====",e);
			collector.fail(tuple);
			throw new FailedException(e);
		}	
	}

	private void emitCountingData(OutputCollector collector) {

		HBaseRecordAdd hbaseInstance = HBaseRecordAdd.getInstance(
				zooserver);   

		String rowkeyAbnArea = null;
		String rowkeyAbnArea2 = null;
		String rowkeyAbnArea3 = null;
		String rowkeyAbnSessH = null;
		String rowKeyAbnSessSourH = null;
		String rowKeyAbnSessSourD = null;
		String rowkeyAbnSessD = null;
		String rowkeyAbnSessD2 = null;


		for(Entry<String,Long> entry : abnAreaMap.entrySet()) {
			//key: dip + "_" + dateStr_d + "_" + sourceAreaId + "_" + areaName + "_" + idcno;
			String key = entry.getKey();
			int abnAreaPV = entry.getValue().intValue();
			String[] splits1 = key.split("\\*",-1);
			rowkeyAbnArea = splits1[0] + "_" + splits1[1] + "_" + splits1[2];
			hbaseInstance.Add("SDS_ABNORMAL_AREA", rowkeyAbnArea, "cf", "IP", splits1[0]);
			hbaseInstance.Add("SDS_ABNORMAL_AREA", rowkeyAbnArea, "cf", "REPORTTIME", splits1[1]);
			hbaseInstance.Add("SDS_ABNORMAL_AREA", rowkeyAbnArea, "cf", "AREA", splits1[3]);
			hbaseInstance.Add("SDS_ABNORMAL_AREA", rowkeyAbnArea, "cf", "COUNTRY", splits1[5]);
			hbaseInstance.incerment("SDS_ABNORMAL_AREA", rowkeyAbnArea, "cf:PV", abnAreaPV);

			rowkeyAbnArea2 = splits1[1] + "_" + splits1[2];
			hbaseInstance.Add("SDS_ABNORMAL_AREA", rowkeyAbnArea2, "cf", "IP", splits1[0]);
			hbaseInstance.Add("SDS_ABNORMAL_AREA", rowkeyAbnArea2, "cf", "REPORTTIME", splits1[1]);
			hbaseInstance.Add("SDS_ABNORMAL_AREA", rowkeyAbnArea2, "cf", "AREA", splits1[3]);
			hbaseInstance.incerment("SDS_ABNORMAL_AREA", rowkeyAbnArea2, "cf:PV", abnAreaPV);

			rowkeyAbnArea3 = splits1[4] + "_" + splits1[1]+ "_" + splits1[2];
			hbaseInstance.Add("SDS_ABNORMAL_AREA", rowkeyAbnArea3, "cf", "IP", splits1[0]);
			hbaseInstance.Add("SDS_ABNORMAL_AREA", rowkeyAbnArea3, "cf", "REPORTTIME", splits1[1]);
			hbaseInstance.Add("SDS_ABNORMAL_AREA", rowkeyAbnArea3, "cf", "AREA", splits1[3]);
			hbaseInstance.incerment("SDS_ABNORMAL_AREA", rowkeyAbnArea3, "cf:PV", abnAreaPV);


		}

		for(Entry<String,Long> entry : abnSessHMap.entrySet()) {
			String key = entry.getKey();
			int abnSessHPV = entry.getValue().intValue();
//			rowkeyAbnSessH = key.substring(0,key.lastIndexOf("_"));
			String[] splits2 = key.split("_",-1);
			rowkeyAbnSessH = splits2[0] + "_" + splits2[1];
			String type = splits2[3];
//			if(splits2.length != 4){
//				System.err.println(">>>>>>>>>>>>>>>>>>>>SDS_ABN_SESS_STAT_H......PV=====" + abnSessHPV + ",key is " + key + ", IP is " + splits2[0] + ", REPORTTIME is " + splits2[1] + ", HOUR is " + splits2[2]);
//			}
			hbaseInstance.Add("SDS_ABN_SESS_STAT_H", rowkeyAbnSessH, "cf", "IP", splits2[0]);
			hbaseInstance.Add("SDS_ABN_SESS_STAT_H", rowkeyAbnSessH, "cf", "REPORTTIME", splits2[1]);
			hbaseInstance.Add("SDS_ABN_SESS_STAT_H", rowkeyAbnSessH, "cf", "HOUR", splits2[2]);
			hbaseInstance.incerment("SDS_ABN_SESS_STAT_H", rowkeyAbnSessH, "cf:"+type, abnSessHPV);

		}

		for(Entry<String,Long> entry : abnSessDPacketMap.entrySet()) {
			String key = entry.getKey();
			int abnSessDPV = entry.getValue().intValue();
			String[] splits3 = key.split("\\*",-1);
			rowkeyAbnSessD = splits3[0] + "_" + splits3[1];
			String type = splits3[2];
			hbaseInstance.Add("SDS_ABN_SESS_STAT_D", rowkeyAbnSessD, "cf", "IP", splits3[0]);
			hbaseInstance.Add("SDS_ABN_SESS_STAT_D", rowkeyAbnSessD, "cf", "REPORTTIME", splits3[1]);
			hbaseInstance.incerment("SDS_ABN_SESS_STAT_D", rowkeyAbnSessD, "cf:"+type, abnSessDPV);
			hbaseInstance.incerment("SDS_ABN_SESS_STAT_D", rowkeyAbnSessD, "cf:ATTOCT", abnSessDOctetsMap.get(key));
			hbaseInstance.incerment("SDS_ABN_SESS_STAT_D", rowkeyAbnSessD, "cf:ATTPACKET", abnSessDPV);

			System.err.println("....SDS_ABN_SESS_STAT_D......key=" + key);
			rowkeyAbnSessD2 = splits3[1] + "_" + splits3[3] + "_" + splits3[0];
			hbaseInstance.Add("SDS_ABN_SESS_STAT_IDC_D", rowkeyAbnSessD2, "cf", "IP", splits3[0]);
			hbaseInstance.Add("SDS_ABN_SESS_STAT_IDC_D", rowkeyAbnSessD2, "cf", "REPORTTIME", splits3[1]);
			hbaseInstance.Add("SDS_ABN_SESS_STAT_IDC_D", rowkeyAbnSessD2, "cf", "IDC", splits3[3]);
			hbaseInstance.incerment("SDS_ABN_SESS_STAT_IDC_D", rowkeyAbnSessD2, "cf:"+type, abnSessDPV);
		}

		//		for(Entry<String,Integer> entry : abnSessSourHMap.entrySet()) {
		//			String key = entry.getKey();
		//			int abnSourHPV = entry.getValue().intValue();
		//			rowKeyAbnSessSourH =  key.substring(0,key.lastIndexOf("_"));
		//			String[] splits4 = key.split("_");
		//			if(splits4.length != 4) {   
		//				System.err.println(">>>>>>>>>>>>>>>>>>>>>>>>>>>>SDS_ABN_SESS_SOUR_STAT_H......PV=====" + abnSourHPV + ",key is " + key + ", DESTIP is " + splits4[0] + ", SOURCEIP is " + splits4[2] + ", SOURCEAREA is " + splits4[3]);
		//			}
		//			hbaseInstance.Add("SDS_ABN_SESS_SOUR_STAT_H", rowKeyAbnSessSourH, "cf", "DESTIP", splits4[0]);
		//			hbaseInstance.Add("SDS_ABN_SESS_SOUR_STAT_H", rowKeyAbnSessSourH, "cf", "SOURCEIP", splits4[2]);
		//			hbaseInstance.Add("SDS_ABN_SESS_SOUR_STAT_H", rowKeyAbnSessSourH, "cf", "SOURCEAREA", splits4[3]);
		//			hbaseInstance.incerment("SDS_ABN_SESS_SOUR_STAT_H", rowKeyAbnSessSourH, "cf:PV", abnSourHPV);
		//		}

		//		for(Entry<String,Integer> entry : abnSessSourDMap.entrySet()) {
		//			String key = entry.getKey();
		//			int abnSourDPV = entry.getValue().intValue();
		//			rowKeyAbnSessSourD =  key.substring(0,key.lastIndexOf("_"));
		//			String[] splits5 = key.split("_",-1);
		//			//			System.err.println("SDS_ABN_SESS_SOUR_STAT_D......PV=====" + abnSourDPV + "key is " + key);
		//			//			hbaseInstance.Add("SDS_ABN_SESS_SOUR_STAT_D", rowKeyAbnSessSourD, "cf", "DESTIP", splits5[0]);
		//			//			hbaseInstance.Add("SDS_ABN_SESS_SOUR_STAT_D", rowKeyAbnSessSourD, "cf", "REPORTTIME", splits5[1]);
		//			//			hbaseInstance.Add("SDS_ABN_SESS_SOUR_STAT_D", rowKeyAbnSessSourD, "cf", "SOURCEIP", splits5[2]);
		//			//			hbaseInstance.Add("SDS_ABN_SESS_SOUR_STAT_D", rowKeyAbnSessSourD, "cf", "SOURCEAREA", splits5[3]);
		//			//			hbaseInstance.incerment("SDS_ABN_SESS_SOUR_STAT_D", rowKeyAbnSessSourD, "cf:PV", abnSourDPV);
		////			collector.emit("DELAY_", new Values(splits5[0],splits5[2],splits5[3],abnSourDPV));
		//		}

		SimpleDateFormat sdfm = new SimpleDateFormat("mm");
		SimpleDateFormat sdfh = new SimpleDateFormat("yyyyMMddHH00");
		SimpleDateFormat sdfh_ = new SimpleDateFormat("yyyyMMddHH");
		SimpleDateFormat sdfd = new SimpleDateFormat("yyyyMMdd");
		Date date = new Date();
		String dateStrM = sdfm.format(date);
		String dateStrH = sdfh.format(date);
		String dateStrH_ = sdfh_.format(date);
		String dateStrD = sdfd.format(date);
		int nMin5 = (Integer.parseInt(dateStrM)/5)*5;
		int nMin30 = (Integer.parseInt(dateStrM)/30)*30;
		String sMin5 = String.format("%02d", nMin5);
		String sMin30 = String.format("%02d", nMin30);
		String dateStr5 = dateStrH_+sMin5;
		String dateStr30 = dateStrH_+sMin30;
		String rowKeyAbnSession5Min = null;
		String rowKeyAbnSession30Min = null;
		String rowKeyAbnSessionH = null;
		String rowKeyAbnSessionD = null;
		//		String rowKeyAbnSdsD = dateStrD;

		//		for(Entry<String,Integer> entry : abnSessDMap.entrySet()) {
		//			String key = entry.getKey();
		//			int typePV = entry.getValue().intValue();
		//			String[] splits3 = key.split("\\*",-1);
		//			rowKeyAbnSession5Min = splits3[0] + "_" + dateStr5;
		//			rowKeyAbnSession30Min = splits3[0] + "_" + dateStr30;
		//			rowKeyAbnSessionH = splits3[0] + "_" + dateStrH;
		//			rowKeyAbnSessionD = splits3[0] + "_" + dateStrD;
		//			String type = splits3[2];
		//			String idcno = splits3[3];
		//			String dip = splits3[0];
		//			System.out.println("AbnStatis............type: " + type + " ,idcno=" + idcno +",  key=" + key);
		//			//			hbaseInstance.incerment("SDS_SESSION_STAT_H", rowKeyAbnSession5Min, "cf:"+type, typePV);
		////			hbaseInstance.Add("SDS_SESSION_STAT_5MIN", rowKeyAbnSession5Min, "cf", type, (long)typePV);
		//
		//			//--------待屏蔽 ---- Modified  by  turk 2016-04-05
		//			//			hbaseInstance.Add("SDS_SESSION_STAT_H", rowKeyAbnSession5Min, "cf", type, (long)typePV);
		//
		//			//			hbaseInstance.incerment("SDS_SESSION_STAT_30MIN", rowKeyAbnSession30Min, "cf:"+type, typePV);
		//			//			hbaseInstance.incerment("SDS_SESSION_STAT_HOUR", rowKeyAbnSessionH, "cf:"+type, typePV);
		//			//			hbaseInstance.incerment("SDS_SESSION_STAT_DAY", rowKeyAbnSessionD, "cf:"+type, typePV);
		//			////			hbaseInstance.incerment("SDS_IDC_STAT_D", rowKeyAbnSdsD, "cf:"+type, typePV);
		////			collector.emit(new Values(idcno,type,typePV));
		////			collector.emit("DELAY",new Values(idcno,dip,type,typePV));
		//		}

		abnAreaMap.clear();
		abnSessHMap.clear();
		abnSessDOctetsMap.clear();
		abnSessDPacketMap.clear();
		abnSessSourHMap.clear();
		abnSessSourDMap.clear();
		abnSessionPVMap.clear();
	}

	private void countInLocal(Tuple tuple) {

		if(tuple != null ) {  
			//			System.err.println("..........AbnStatisBolt......... " + System.currentTimeMillis());
			String dip = tuple.getStringByField("dip");
			String sip = tuple.getStringByField("sip");
			String idcno = tuple.getStringByField("idcno");
			//			dport = tuple.getStringByField("dport");
			//			sport = tuple.getStringByField("sport");
			String dateStr_d = tuple.getStringByField("dateStr_d"); 
			String dateStr_h = tuple.getStringByField("dateStr_h");
			String dateStr_hour = tuple.getStringByField("dateStr_hour");
			String sourceAreaId = tuple.getStringByField("sourceAreaId");
			String areaName = tuple.getStringByField("areaName");
			String sourceAreaName = tuple.getStringByField("sourceAreaName");
			long PV = tuple.getLongByField("pv");
//			triplePV = tuple.getLongByField("pv");
			long triplePV = tuple.getLongByField("streampacket");
			String abnormaltype = tuple.getStringByField("abntype");
			String sourceCountry = tuple.getStringByField("sourcecountry");
			long streamOctets = tuple.getLongByField("streamoctets");
			long streamPacket = tuple.getLongByField("streampacket");
			//			if(abnormaltype == null || "".equals(abnormaltype)) {
			//				abnormaltype = "SESSIONMIDDLE";
			//			}
			String abnAreaMapKey = dip + "*" + dateStr_d + "*" + sourceAreaId + "*" + areaName + "*" + idcno + "*" + sourceCountry;
			String abnSessHMapKey = dip + "_" + dateStr_h + "_" + dateStr_hour + "_" + abnormaltype;
			String abnSessDMapKey = dip + "*" + dateStr_d + "*" + abnormaltype + "*" + idcno;
			String abnSessSourHMapKey = dip + "_" + dateStr_h + "_" + sip + "_" + sourceAreaName;
			String abnSessSourDMapKey = dip + "_" + dateStr_d + "_" + sip + "_" + sourceAreaName;

			Long abnAreaPV = triplePV;
			Long abnSessHPV = triplePV;
			Long abnSessDPV = triplePV;
			Long abnSourHPV = triplePV;
			Long abnSourDPV = triplePV; 
			Long abnSessionPV = triplePV;
			Long abnSessOctets = streamOctets;

			if(abnAreaMap.get(abnAreaMapKey) != null) {
				abnAreaPV = abnAreaMap.get(abnAreaMapKey) + triplePV;
			}
			if(abnSessHMap.get(abnSessHMapKey) != null) {
				abnSessHPV = abnSessHMap.get(abnSessHMapKey) + triplePV;
			}
			if(abnSessDPacketMap.get(abnSessDMapKey) != null) {
				abnSessDPV = abnSessDPacketMap.get(abnSessDMapKey) + triplePV;
			}            
			if(abnSessDOctetsMap.get(abnSessDMapKey) != null) {
				abnSessOctets = abnSessDOctetsMap.get(abnSessDMapKey) + abnSessOctets;
			}            
			if(abnSessSourHMap.get(abnSessSourHMapKey) != null) {
				abnSourHPV = abnSessSourHMap.get(abnSessSourHMapKey) + triplePV;
			}
			if(abnSessSourDMap.get(abnSessSourDMapKey) != null) {
				abnSourDPV = abnSessSourDMap.get(abnSessSourDMapKey) + triplePV;
			}
			if(abnSessionPVMap.get(dip) != null) {
				abnSessionPV = abnSessionPVMap.get(dip) + PV;
			}

			abnAreaMap.put(abnAreaMapKey, abnAreaPV);
			abnSessHMap.put(abnSessHMapKey, abnSessHPV);
			abnSessDPacketMap.put(abnSessDMapKey, abnSessDPV);
			abnSessDOctetsMap.put(abnSessDMapKey, abnSessOctets);
			abnSessSourHMap.put(abnSessSourHMapKey, abnSourHPV);
			abnSessSourDMap.put(abnSessSourDMapKey, abnSourDPV);
			abnSessionPVMap.put(dip, abnSessionPV);

			
				if("PORTLOW".equals(abnormaltype)) {
//					collector.emit("DELAY_", new Values(dip,sip,areaName,abnSourDPV,sourceCountry,streamOctets,streamPacket,"1"));
				} else {
//					if(streamOctets >0 && streamPacket > 0) {
						collector.emit("DELAY_", new Values(dip,sip,areaName,abnSourDPV,sourceCountry,streamOctets,streamPacket,"0"));
//					} else {
//						collector.emit("DELAY_", new Values(dip,sip,areaName,abnSourDPV,sourceCountry,0L,0L,""));
//					}
				}
			
			collector.emit(new Values(idcno,abnormaltype,triplePV));
			
			//For 5/30/hour/day min stat.
			collector.emit("DELAY",new Values(idcno,dip,abnormaltype,streamOctets,streamPacket,PV));
		} 
	}

	@Override
	public void cleanup() {
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("IDC","TYPE","PV"));
		
		//For 5/30/hour/day min stat.
		declarer.declareStream("DELAY", new Fields("IDC","IP","TYPE","OCTETS","PACKET","PV"));
		
		declarer.declareStream("DELAY_", new Fields("DIP","SIP","AREANAME","PV","SOURCECOUNTRY","STREAMOCTETS","STREAMPACKET","ABNTYPE"));
	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		Map<String,Object> conf = new HashMap<String,Object>();
		conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS,900);
		return conf;
	}

	@Override
	public void Init(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		// TODO Auto-generated method stub
		
	}

}
